/** To projec out the required attributes from the result **/

package qp.operators;

import qp.utils.*;

import java.util.Vector;

public class Project extends Operator
{

	Operator base;
	Vector attrSet; 
	
	int batchsize; // number of tuples per outbatch

	boolean eos;
	int cur;
	/**
	 * The following fields are requied during execution of the Project Operator
	 **/

	Batch inbatch;
	Batch outbatch;

	/**
	 * index of the attributes in the base operator that are to be projected
	 **/

	int[] attrIndex;

	public Project(Operator base, Vector as, int type)
	{
		super(type);
		this.base = base;
		this.attrSet = as;

	}

	public void setBase(Operator base)
	{
		this.base = base;
	}

	public Operator getBase()
	{
		return base;
	}

	public Vector getProjAttr()
	{
		return attrSet;
	}

	/**
	 * Opens the connection to the base operator Also figures out what are the
	 * columns to be projected from the base operator
	 **/

	public boolean open()
	{
		eos = false;
		cur = 0;
		/** setnumber of tuples per batch **/
		int tuplesize = schema.getTupleSize();
		batchsize = Batch.getPageSize() / tuplesize;

		/**
		 * The followingl loop findouts the index of the columns that are
		 * required from the base operator
		 **/

		Schema baseSchema = base.getSchema();
		
		attrIndex = new int[attrSet.size()];
		//System.out.println("Project---Schema: ----------in open-----------");
		//System.out.println("base Schema---------------");
		//Debug.PPrint(baseSchema);
		for (int i = 0; i < attrSet.size(); i++)
		{
			Attribute attr = (Attribute) attrSet.elementAt(i);
			int index = baseSchema.indexOf(attr);
			attrIndex[i] = index;
		}

		if (base.open())
			return true;
		else
			return false;
	}

	/** Read next tuple from operator */

	public Batch next()
	{
		int i = 0;
		if(eos)
			return null;
		outbatch = new Batch(batchsize);
		
		/**
		 * keep on checking the incoming pages until the output buffer is full
		 **/
		while (!outbatch.isFull())
		{
			if (cur == 0)
			{// the first inbatch or already read the last inbatch
				inbatch = base.next();
				/** There is no more incoming pages from base operator **/
				if (inbatch == null)
				{

					eos = true;
					return outbatch;
				}
			}

			/**
			 * Continue this for loop until this page is fully observed or the
			 * output buffer is full
			 **/

			for (i = cur; i < inbatch.size() && (!outbatch.isFull()); i++)
			{
				Tuple basetuple = inbatch.elementAt(i);
				//Debug.PPrint(basetuple);
				//System.out.println();
				Vector present = new Vector();
				for (int j = 0; j < attrSet.size(); j++)
				{
					Object data = basetuple.dataAt(attrIndex[j]);
					present.add(data);
				}
				Tuple outtuple = new Tuple(present);
				outbatch.add(outtuple);
				
			}

			/**
			 * Modify the cursor to the position requierd when the base operator
			 * is called next time;
			 **/

			if (i == inbatch.size())
				cur = 0;
			else
				cur = i;

			// return outbatch;
		}
		return outbatch;
		
		/*//System.out.println("Project:-----------------in next-----------------");
		outbatch = new Batch(batchsize);

		

		inbatch = base.next();
		// System.out.println("Project:-------------- inside the next---------------");

		if (inbatch == null)
		{
			return null;
		}
		//System.out.println("Project:---------------base tuples---------");
		for (int i = 0; i < inbatch.size(); i++)
		{
			Tuple basetuple = inbatch.elementAt(i);
			//Debug.PPrint(basetuple);
			//System.out.println();
			Vector present = new Vector();
			for (int j = 0; j < attrSet.size(); j++)
			{
				Object data = basetuple.dataAt(attrIndex[j]);
				present.add(data);
			}
			Tuple outtuple = new Tuple(present);
			outbatch.add(outtuple);
		}
		return outbatch;*/
	}

	/** Close the operator */
	public boolean close()
	{
		return true;
		/*
		 * if(base.close()) return true; else return false;
		 */
	}

	public Object clone()
	{
		Operator newbase = (Operator) base.clone();
		Vector newattr = new Vector();
		for (int i = 0; i < attrSet.size(); i++)
			newattr.add((Attribute) ((Attribute) attrSet.elementAt(i)).clone());
		Project newproj = new Project(newbase, newattr, optype);
		Schema newSchema = newbase.getSchema().subSchema(newattr);
		newproj.setSchema(newSchema);
		return newproj;
	}
}
